package com.grf.kafka.java.simple;

import ch.qos.logback.classic.Logger;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/**
 * @apiNote 消费者
 * @author grf
 * @since 2024/4/4
 */
@Slf4j
public class SimpleConsumer {
    private static final String BOOTSTRAP_SERVERS = "127.0.0.1:9092";

    private static final String TOPIC = "disTopic";

    public static void main(String[] args) {
        // Part1: 设置发送相关属性
        Properties properties = new Properties();
        // 设置kafka地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        // 每个消费者要指定一个group
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        // 黄色至key序列化化雷
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // 设置value的序列值
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        Consumer<String, String> consumer = new KafkaConsumer(properties);
        consumer.subscribe(Arrays.asList(TOPIC));
        while (true){
            // PART2:拉取消息
            // 100毫秒超时时间
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofNanos(100));

            // PART3: 处理消息
            for(ConsumerRecord<String, String> record : records){
                log.info("offset={};key={};value={}",record.offset(),record.key(),record.value());
            }
            // 提交offset，消息就不会重复推送
            consumer.commitSync();//同步提交，表示必须等到offset提交后再去拉取下一批消息
            // consumer.commitAsync();//异步提交，表示发送完offset的提交请求后，就开始消费下一批数据了。不用等到broker的确认
        }
    }
    /**
     * @description: 禁用debug日志
     * @param: null
     * @return: null
     **/
    static {
        ch.qos.logback.classic.LoggerContext loggerContext = (ch.qos.logback.classic.LoggerContext) org.slf4j.LoggerFactory.getILoggerFactory();
        List<Logger> loggerList = loggerContext.getLoggerList();
        loggerList.forEach(logger -> logger.setLevel(ch.qos.logback.classic.Level.INFO));
    }
}
